package org.funtester.performance.books.chapter04.section4;

import com.alibaba.fastjson.JSONObject;
import org.apache.http.HttpStatus;
import org.apache.http.client.methods.CloseableHttpResponse;
import org.apache.http.client.methods.HttpRequestBase;
import org.apache.http.impl.client.CloseableHttpClient;
import org.funtester.performance.books.chapter04.section2.HttpClientTool;
import org.funtester.performance.books.chapter04.section2.HttpRequestTool;

import java.io.IOException;
import java.util.List;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * 日志回放类
 */
public class ReplayLog {

    /**
     * token文件路径,默认全局变量
     */
    private static String tokenPath = "token.txt";

    private static String HOST = "http://localhost:8080";

    /**
     * 全局 HttpClient 对象
     */
    public CloseableHttpClient httpClient;

    /**
     * 全局线程池对象
     */
    public ThreadPoolExecutor executor;

    /**
     * 日志队列
     */
    public DelayQueue<FunLog> queue;

    /**
     * 日志文件路径
     */
    public String logPath;

    /**
     * 压测倍数
     */
    public int multiple;

    /**
     * 压测名称
     */
    public String name;

    /**
     * 测试时间,单位秒
     */
    public int testTime;

    /**
     * 配置数据,用于覆盖日志中的参数
     */
    JSONObject configData;

    public ReplayLog(String logPath, int multiple, String name, JSONObject configData,int testTime) {
        this.logPath = logPath;
        this.multiple = multiple;
        this.name = name;
        this.configData = configData;
        this.testTime = testTime;
        this.queue = new DelayQueue<>();// 初始化日志队列
        httpClient = HttpClientTool.getHttpClient();// 获取全局 HttpClient 对象
        executor = new ThreadPoolExecutor(2, 200, 60L, TimeUnit.SECONDS, new SynchronousQueue<>(), new ThreadFactory() {// 初始化线程池对象

            AtomicInteger index = new AtomicInteger();// 线程索引

            @Override
            public Thread newThread(Runnable r) {
                Thread thread = new Thread(r);// 创建线程
                thread.setName("ReplayLog-" + index.incrementAndGet());// 设置线程名称
                return thread;
            }
        }, new RejectedExecutionHandler() {
            @Override
            public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
                System.out.println("线程池已经满载,任务被拒绝执行");// 打印线程池已经满载信息
            }
        });
    }


    /**
     * 开始日志回放测试
     */
    public void start() {
        List<String> toknes = FunLogUtils.readLog(tokenPath);// 读取 token 文件,获取所有用户 token
        FunLogUtils funLogUtils = new FunLogUtils(this.logPath);// 创建日志工具类对象
        funLogUtils.readLogs();// 读取日志文件
        long start = System.currentTimeMillis();
        // 计算时间差，并且预留10秒，作为日志读取耗时的buffer，防止因为日志读取延迟导致压测开始流量不均匀
        long diff = start - funLogUtils.minTimestamp + 1 * 1000;
        new Thread(() -> {
            funLogUtils.logs.forEach(log -> {
                log.timestamp = log.timestamp + diff;// 计算时间差,并重新赋值
                queue.offer(log);// 添加到日志队列
            });
        }).start();
        long startTime = System.currentTimeMillis();
        System.out.println(startTime + "  日志回放: " + this.name + " 开始测试");// 打印开始回放测试信息
        AtomicInteger userIndex = new AtomicInteger();// 用户索引,方便并发时顺序获取用户token
        while (true) {
            if (System.currentTimeMillis() - startTime > this.testTime * 1000) {// 判断是否到达测试时间
                break;// 结束测试
            }
            try {
                FunLog funLog = queue.poll(1, TimeUnit.SECONDS);
                if (funLog != null) {
                    for (int i = 0; i < this.multiple; i++) {
                        executor.execute(() -> {
                            String user = toknes.get(userIndex.getAndIncrement() % toknes.size());
                            HttpRequestBase request = funLog.toRequest(HOST, user, this.configData);
                            try (CloseableHttpResponse response = httpClient.execute(request)) {// 执行请求
                                int statusCode = response.getStatusLine().getStatusCode();// 获取响应状态码
                                JSONObject reponseJson = HttpRequestTool.parseReponseJson(response);// 解析响应 JSON
                                if (statusCode == HttpStatus.SC_OK && reponseJson.getInteger("code") == 0) {// 判断响应状态码和响应code
                                    System.out.println(System.currentTimeMillis() + "  " + Thread.currentThread().getName() + "  请求成功,状态码:" + statusCode + ",响应:" + reponseJson);// 打印请求成功信息
                                } else {
                                    System.out.println(System.currentTimeMillis() + "  " + Thread.currentThread().getName() + "  请求失败,状态码:" + statusCode + ",响应:" + reponseJson);// 打印请求失败信息
                                }
                            } catch (IOException e) {
                                e.printStackTrace();
                            }
                        });
                    }
                    funLog.timestamp = funLog.timestamp + funLogUtils.nextDelay;
                    queue.offer(funLog);
                }
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
        executor.shutdown();// 关闭线程池
        System.out.println(System.currentTimeMillis() + "  日志回放: " + this.name + " 结束测试");// 打印结束测试信息
    }


}
